package com.zhisheng.sql.blink.stream.example;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class UserPurchaseSetProcessor extends KeyedProcessFunction<Integer, UserPurchaseSetProcessor.OrderEvent, String> {
    public static Map<Integer,Set<Integer>> userPurchaseMap = new ConcurrentHashMap<>();
    static {
        userPurchaseMap.put(1,new HashSet<>());
        userPurchaseMap.put(2,new HashSet<>());
        userPurchaseMap.put(3,new HashSet<>());
        userPurchaseMap.put(4,new HashSet<>());
        userPurchaseMap.put(5,new HashSet<>());
        userPurchaseMap.put(6,new HashSet<>());
    }
    @Override
    public void open(Configuration parameters) throws Exception {


    }

    @Override
    public void processElement(OrderEvent order, Context context, Collector<String> collector) throws Exception {
        Integer userId = order.getUserId();
        Integer productId = order.getProductId();

        // 获取用户的购买集合，如果不存在则创建新的
        Set<Integer> purchaseSet = userPurchaseMap.get(userId);
        if (purchaseSet == null) {
            purchaseSet = new HashSet<>();
        }

        // 添加新购买的产品
        purchaseSet.add(productId);
        userPurchaseMap.put(userId, purchaseSet);

        // 输出当前用户的所有购买记录
        collector.collect(String.format("User %d has purchased products: %s", userId, purchaseSet));
    }


    // 订单事件类
    public static class OrderEvent {
        private Integer userId;
        private Integer productId;
        private Double amount;

        public OrderEvent(Integer userId, Integer productId, Double amount) {
            this.userId = userId;
            this.productId = productId;
            this.amount = amount;
        }

        public Integer getUserId() {
            return userId;
        }

        public Integer getProductId() {
            return productId;
        }

        public Double getAmount() {
            return amount;
        }
    }
}
